package com.mzj.solr;

import java.util.ArrayList;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.log4j.Logger;
import org.apache.solr.client.solrj.SolrClient;
import org.apache.solr.client.solrj.SolrQuery;
import org.apache.solr.client.solrj.SolrQuery.ORDER;
import org.apache.solr.client.solrj.response.QueryResponse;
import org.apache.solr.common.SolrDocument;
import org.apache.solr.common.SolrDocumentList;
import org.apache.solr.common.SolrInputDocument;
import org.apache.solr.common.params.CursorMarkParams;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.unit.TimeValue;

import com.mzj.util.PropertiesUtils;

public class SolrCursorHbMove {

	private static final Logger LOG = Logger.getLogger(SolrMove.class);  
	
	public static Map<String, String> paramMap = null;
	
	static {
		paramMap = PropertiesUtils.getParamMap();
	}
	
	private static SolrQuery setRequest(SolrQuery query) {
		String start = paramMap.get("hb.start.time");
		String end = paramMap.get("hb.end.time");
		query.setFilterQueries("req_time:["+start+" TO "+end+"]");
		return query;
	}
	
	public static Long getTotal(SolrClient solrClient, String indexName) {
		SolrQuery query = new SolrQuery("*:*");  
		query = setRequest(query);
		query.setRows(1);
		query.setStart(0);
        QueryResponse resp;
		try {
			resp = solrClient.query(query);
			return resp.getResults().getNumFound();
		} catch (Exception e) {
			LOG.error(e.getMessage(),e);
		}  
		return 0L;
	}
	
	public static void move(SolrClient client ,RestHighLevelClient esClient, String indexName,Integer pageSize, 
			String index,Integer start) throws Exception {
		SolrQuery query = new SolrQuery("*:*");  
		query = setRequest(query);
		query.setRows(pageSize);
		query.setStart(start);
		LOG.info("--------------query: start,"+start+" end,"+(start+pageSize));
		//query.addSort("create_time", ORDER.asc);
        QueryResponse resp = client.query(query);  
        
        LOG.info("--------------query:result.....");
        int size = resp.getResults().size();
        SolrDocumentList list = resp.getResults();
        BulkRequest request = new BulkRequest(); 
        for(int i=0; i<size; i++) {
        	SolrDocument doc = list.get(i);
        	Collection<String> ks = doc.getFieldNames();
        	Map<String,Object> input = new HashMap<String,Object>();
        	for (String k : ks) {
        		if(!"_version_".equals(k)){
        			input.put(k, doc.getFieldValue(k));
        		}
        	}  
        	request.add(new IndexRequest(indexName+"2", "data",input.get("id").toString()).source(input));
        }
        LOG.info("--------------submit:");
        request.timeout(TimeValue.timeValueMinutes(60));
        
        BulkResponse bulk = esClient.bulk(request);
		
		if(bulk.hasFailures()){
			bulk = esClient.bulk(request);
			if(bulk.hasFailures()){
				LOG.info("====================================ES data has submited fail! "+start );
				LOG.info("====================================ES data has submited fail! "+bulk.buildFailureMessage());
			}
		}else {
			LOG.info("====================================ES data has submited sucess! "+start );
		}
	}
	
	
	// 游标查询
	public static void cursorQuery(SolrClient client ,RestHighLevelClient esClient, String indexName,Integer size)  {
		// solr查询封装
		SolrQuery sq = new SolrQuery();
		sq.setRows(size);// 设置游标一次读的数量
		sq.set("q", "*:*");// 按条件检索
		sq.setFilterQueries("req_time:[20180103000000 TO 20180104095959]");
		sq.setSort("id", ORDER.asc);// 根据主键排序
		
		String cursorMark = CursorMarkParams.CURSOR_MARK_START;// 游标初始化
		boolean done = false;
		int i =0;
		BulkRequest request = new BulkRequest(); 
		Map<String,Object> input = null;
		int start = i;
		while (!done) {
			try {
				sq.set(CursorMarkParams.CURSOR_MARK_PARAM, cursorMark);// 变化游标条件
				QueryResponse rsp = client.query(sq);
				// 执行多次查询读取
				String nextCursorMark = rsp.getNextCursorMark();// 获取下次游标
				// 做一些操作数据的事
				
				for (SolrDocument sd : rsp.getResults()) {
					String id = (String) sd.get("id");
					
					SolrDocument doc = client.getById(id);
					Collection<String> ks = doc.getFieldNames();
		        	input = new HashMap<String,Object>();
		        	for (String k : ks) {
		        		if(!"_version_".equals(k)){
		        			input.put(k, doc.getFieldValue(k));
		        		}
		        	}  
		        	request.add(new IndexRequest(indexName+"1", "data",input.get("id").toString()).source(input));
		        	i++;
				}
				request.timeout(TimeValue.timeValueMinutes(10));
				BulkResponse bulk = esClient.bulk(request);
				
				if(bulk.hasFailures()){
					bulk = esClient.bulk(request);
					if(bulk.hasFailures()){
						LOG.info("====================================ES data has submited fail! "+start +":"+i );
						LOG.info("====================================ES data has submited fail! "+bulk.buildFailureMessage());
					}
				}else {
					LOG.info("====================================ES data has submited sucess! "+start +":"+i );
				}
				start = i;
			
				// 如果两次游标一样，说明数据拉取完毕，可以结束循环了
				if (cursorMark.equals(nextCursorMark)) {
					done = true;
				}
				cursorMark = nextCursorMark;
				
			} catch (Exception e) {
				LOG.error(e);
			}
			System.gc();
		}
	}
	
	// 游标查询
	public static void cursorQuery(SolrClient client ,SolrClient cloudClient, String indexName) throws Exception {
		// solr查询封装
		SolrQuery sq = new SolrQuery();
		sq.setRows(10000);// 设置游标一次读的数量
		sq.set("q", "*:*");// 按条件检索
		sq.setFilterQueries("req_time:[20171226 TO 20171227]");
		sq.setSort("id", ORDER.asc);// 根据主键排序
		String cursorMark = CursorMarkParams.CURSOR_MARK_START;// 游标初始化
		boolean done = false;
		List<SolrInputDocument> inpList = null;
		int i =1;
		while (!done) {
			sq.set(CursorMarkParams.CURSOR_MARK_PARAM, cursorMark);// 变化游标条件
			QueryResponse rsp = client.query(sq);// 执行多次查询读取
			String nextCursorMark = rsp.getNextCursorMark();// 获取下次游标
			// 做一些操作数据的事
			inpList = new ArrayList<SolrInputDocument>();
			
			for (SolrDocument sd : rsp.getResults()) {
				String id = (String) sd.get("id");
				LOG.info("--------------cursorQuery:"+i+", id:"+id);
				
				SolrDocument doc = client.getById(id);
				Collection<String> ks = doc.getFieldNames();
	        	SolrInputDocument input = new SolrInputDocument();
	        	for (String k : ks) {
	        		if(!"_version_".equals(k)){
	        			input.addField(k, doc.getFieldValue(k));
	        		}
	        	}  
	        	inpList.add(input);
	        	i++;
			}
	        
			cloudClient.add(indexName,inpList);
	        cloudClient.commit(indexName);
			// 如果两次游标一样，说明数据拉取完毕，可以结束循环了
			if (cursorMark.equals(nextCursorMark)) {
				done = true;
			}
			cursorMark = nextCursorMark;
		}
	}



	
}
